package main

import "crypto/tls"
import "fmt"
import "log"
import "time"
import "github.com/GaryBoone/GoStats/stats"
import mqtt "github.com/eclipse/paho.mqtt.golang"

type Client struct {
	ID int
	ClientId string
	BrokerURL string
	BrokerUser string
	BrokerPass string
	MsgTopic  string
	MsgPayload string
	MsgSize   int
	MsgCount  int
	MsgQos   byte
	Quiet   bool
	WaitTimeout  time.Duration
	TLSConfig  *tls.Config
	MessageInterval int
}

func (c *Client) Run(res chan *RunResults) {
	newMsgs := make(chan *Message)
	pubMsgs := make(chan *Message)
	doneGen := make(chan bool)
	donePub := make(chan bool)
	runResults := new(RunResults)

	started := time.Now()
	go c.genMessages(newMsgs, doneGen)
	go c.pubMessages(newMsgs, pubMsgs, doneGen, donePub)

	runResults.ID = c.ID
	times := []float64{}
	for{
		select {
		case m:=<-pubMsgs:
			if m.Error {
				log.Printf("Client %v ERROR publishing message: %v: at%v\n", c.ID, m.Topic, m.Sent.Unix())
				runResults.Failures++
			}else{
				runResults.Successes++
				times = append(times, m.Delivered.Sub(m.Sent).Seconds() * 1000)
			}
		case <- donePub:
			duration := time.Since(started)
			runResults.MsgTimeMin = stats.StatsMin(times)
			runResults.MsgTimeMax = stats.StatsMax(times)
			runResults.MsgTimeMean = stats.StatsMean(times)
			runResults.RunTime = duration.Seconds()
			runResults.MsgPerSec = float64(runResults.Successes)/duration.Seconds()
			if c.MsgCount > 1{
				runResults.MsgTimeStd = stats.StatsSampleStandardDeviation(times)
			}

			res <- runResults
			return
		}
	}
}

func (c *Client) genMessages(ch chan *Message, done chan bool)  {
	var payload interface{}

	if c.MsgPayload != ""{
		payload = c.MsgPayload
	}else {
		payload = make([]byte, c.MsgSize)
	}

	for i := 0; i < c.MsgCount; i++ {
		ch <- &Message{
			Topic: c.MsgTopic,
			Qos: c.MsgQos,
			Payload: payload,
		}
		time.Sleep(time.Duration(c.MessageInterval) * time.Second)
	}
	done <- true
}

func (c *Client) pubMessages(in, out chan *Message, doneGen, donePub chan bool) {
	onConnected := func(client mqtt.Client) {
		if !c.Quiet{
			log.Printf("Client %v is connected to the broken %v\n", c.ID, c.BrokerURL)
		}

		ctr := 0
		for{
			select {
			case m:= <- in:
				m.Sent = time.Now()
				token := client.Publish(m.Topic, m.Qos, false, m.Payload)
				res := token.WaitTimeout(c.WaitTimeout)
				if !res{
					log.Printf("Client %v Timeout Sending message: %v\n", c.ID, token.Error())
					m.Error = true
				}else if token.Error() != nil{
					log.Printf("Client %v Error sending message: %v\n", c.ID, token.Error())
					m.Error = true
				}else {
					m.Delivered = time.Now()
					m.Error = false
				}
				out <- m

				if ctr > 0 && ctr % 100 == 0{
					if !c.Quiet{
						log.Printf("Client %v published %v messages and keeps publishing...\n", c.ID, ctr)
					}
				}
				ctr++
			case <- doneGen:
				donePub <- true
				if !c.Quiet{
					log.Printf("Client %v is done publishing\n", c.ID)
				}
				return
			}
		}
	}

	opts := mqtt.NewClientOptions().
			AddBroker(c.BrokerURL).
			SetClientID(fmt.Sprintf("%s-%v", c.ClientId, c.ID)).
			SetCleanSession(true).
			SetAutoReconnect(true).
			SetOnConnectHandler(onConnected).
			SetConnectionLostHandler(func(client mqtt.Client, reason error) {
				log.Printf("Client %v lost connection to the broken: %v. will reconnect...\n", c.ID, reason.Error())
	})
	if c.BrokerUser != "" && c.BrokerPass != "" {
		opts.SetUsername(c.BrokerUser)
		opts.SetPassword(c.BrokerPass)
	}
	if c.TLSConfig != nil {
		opts.SetTLSConfig(c.TLSConfig)
	}

	client := mqtt.NewClient(opts)
	token := client.Connect()
	token.Wait()

	if token.Error() != nil{
		log.Printf("Client %v has error connecting to the broken: %v\n", c.ID, token.Error())
	}
}
